package com.integrate.rabbitmq.porducer;

import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * @author 刘志强
 * @date 2020/8/18 14:47
 */
@Component
@Slf4j
public class AckSender implements  RabbitTemplate.ConfirmCallback , RabbitTemplate.ReturnCallback{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 指定队列发送
     * @param queue
     * @param content
     * @return
     */
    public CorrelationData convertAndSend(String queue, Object content) {
        //设置回调对象
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
        //构建回调返回的数据
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(queue,content,correlationData);
        return correlationData;
    }

    /**
     * 指定交换机 和 路由建 发送
     * @param exchange
     * @param routingKey
     * @param content
     * @return
     */
    public CorrelationData convertAndSend(String exchange, String routingKey, Object content) {
        //设置回调对象
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
        //构建回调返回的数据
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(exchange,routingKey, content,correlationData);
        return correlationData;
    }

    public CorrelationData convertAndSendDelay(String queue, Object content, Long time) {
        //设置回调对象
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
        //构建回调返回的数据
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        MessagePostProcessor messagePostProcessor = new MyMessagePostProcessor(time);
        rabbitTemplate.convertAndSend(queue, content,messagePostProcessor,correlationData);
        return correlationData;
    }

    /**
     * 消息回调确认方法
     * 如果消息没有到exchange,则confirm回调,ack=false
     * 如果消息到达exchange,则confirm回调,ack=true
     * @param
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean isSendSuccess, String s) {
        log.info("confirm--message:回调消息ID为: " + correlationData.getId());
        if (isSendSuccess) {
            log.info("消息进入交换机成功");
        } else {
            log.info("消息进入交换机失败====" + s);
        }
    }

    /**
     * exchange到queue成功,则不回调return
     * exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        StringBuilder str = new StringBuilder();
        str.append("return--message:").append(message.getBody())
                .append(",replyCode:").append(replyCode).append(",replyText:").append(replyText).append(",exchange:").append(exchange)
                .append(",routingKey:").append(routingKey);
        log.info(String.valueOf(str));
    }
}
